package org.wenzi.com.function;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.wenzi.com.pojo.UserBehaviorPojo;

/**
 * @author zhaozuowen
 * @date 2021-10-25 11:41
 */
public class WindowResultFunction implements WindowFunction<Long, WindowResultFunction.ItemViewCount, Tuple, TimeWindow> {

    @Override
    public void apply(Tuple key, TimeWindow timeWindow, Iterable<Long> iterable, Collector<ItemViewCount> collector) throws Exception {
        Long itemId = ((Tuple1<Long>) key).f0;
        Long viewCount = iterable.iterator().next();
        collector.collect(new ItemViewCount(itemId,timeWindow.getStart(),timeWindow.getEnd(),viewCount));

    }


    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class ItemViewCount{
        public long itemId;     // 商品ID
        public long windowStart; //窗口开始时间戳
        public long windowEnd;  // 窗口结束时间戳
        public long viewCount;  // 商品的点击量


    }
}
